﻿using MTP.Buffer;
using MTP.Utils;

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace MTP.Dispatch
{
    class SocketStream : IDisposable
    {
        private Socket socket;
        private readonly byte[] cipher;
        private readonly DispatchBase dispatch;

        private readonly ConcurrentDictionary<Guid, AsyncBufferStream> concurrentDictionary = new ConcurrentDictionary<Guid, AsyncBufferStream>();

        public bool IsClose { get; set; } = false;

        private Guid guid = Guid.NewGuid();
        public string ClientId => guid.ToString("N");

        public bool Connected => !IsClose && socket.Connected;

        public readonly string RemoteHost;

        public SocketStream(Socket socket, byte[] cipher, DispatchBase dispatch)
        {
            RemoteHost = socket?.RemoteEndPoint?.ToString();
            this.socket = socket;
            this.cipher = cipher;
            if (dispatch != null)
            {
                this.dispatch = dispatch;
            }
        }


        /// <summary>
        /// 连接
        /// </summary>
        /// <param name="address"></param>
        /// <param name="port"></param>
        public async Task Connect(Socket socket, EndPoint iPEndPoint, CancellationToken cancellationToken)
        {
            this.socket = socket;
            socket.Connect(iPEndPoint);
            while (!socket.Connected)
            {
                await Task.Yield();
            }
            Write(Util.Handshake(cipher, guid));
            socket.ReceiveTimeout = 2000;
            if (Util.VerifyHandshake(socket.Receive, cipher, out _))
            {
                socket.ReceiveTimeout = 0;
                IsClose = false;
                RunListen(cancellationToken);
                return;
            }
            socket.Dispose();
            throw new IOException("the connection has been disconnected.");
        }

        public void RunListen(CancellationToken cancellationToken)
        {
            Util.LongFactory.StartNew(() =>
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    try
                    {
                        ReadPackage();
                    }
                    catch (Exception)
                    {
                        break;
                    }
                }
                Close();
            }, cancellationToken);
        }

        /// <summary>
        /// 绑定监听
        /// </summary>
        /// <param name="iPAddress"></param>
        /// <param name="port"></param>
        /// <param name="backlog"></param>
        public void Bind(DispatchBase dispatchBase, CancellationToken token, EndPoint endPoint, int backlog = 10)
        {
            if (dispatch != null)
            {
                throw new IOException("Can`t bind");
            }
            socket.Bind(endPoint);
            socket.Listen(backlog);

            Util.LongFactory.StartNew(() =>
            {
                try
                {
                    while (!token.IsCancellationRequested)
                    {
                        Accept(socket.Accept(), dispatchBase);
                    }
                }
                catch
                {
                }
            }, token);
        }

        /// <summary>
        /// 获取新的socket
        /// </summary>
        /// <param name="dispatch"></param>
        /// <returns></returns>
        public void Accept(Socket newsocket, DispatchBase dispatchBase)
        {
            Task.Run(() =>
            {
                try
                {
                    newsocket.ReceiveTimeout = 2000;
                    if (Util.VerifyHandshake(newsocket.Receive, cipher, out guid))
                    {
                        newsocket.ReceiveTimeout = 0;
                        var socketStream = new SocketStream(newsocket, cipher, dispatchBase);
                        socketStream.Write(Util.Handshake(cipher, guid));
                        dispatchBase.Accept(socketStream);
                        return;
                    }
                    // 验证失败
                }
                catch
                {

                }
                newsocket.Close();
                newsocket.Dispose();
            });
        }


        /// <summary>
        /// 
        /// </summary>
        /// <param name="socketStream"></param>
        /// <param name="createAsync"></param>
        /// <returns></returns>
        public void ReadPackage()
        {
            ByteReadBuffer buffer = Read(MTPConst.HEADER_LENGTH);
            var guid = new Guid(buffer.ReadBytes(16));
            var bufferAsync = concurrentDictionary.GetOrAdd(guid, (k) => new AsyncBufferStream());
            // 当前包索引
            int sequence = buffer.ReadInt(2);
            // 包体长度
            int length = buffer.ReadInt(2);
            int index = sequence >> 1;
            bool isEnd = (sequence & 1) == 1;
            if (index == 0)
            {
                PackageType packageType = (PackageType)buffer.ReadByte();
                int headerLength = buffer.ReadInt(2);
                var header = Util.QueryToDictionary(Read(headerLength).ReadEnd());
                if (RemoteHost != null)
                {
                    header.Add("RemoteHost", RemoteHost);
                }
                StartTask(bufferAsync, guid, packageType, header);
            }
            bufferAsync.Append(index, ReadBytes(length), isEnd);
        }

        private void StartTask(AsyncBufferStream asyncBuffer, Guid guid, PackageType packageType, Dictionary<string, string> header)
        {
            Util.TaskFactory.StartNew(() =>
           {
               try
               {
                   dispatch.Receive(guid, new MTPPackage(asyncBuffer, packageType, header), ClientId);
               }
               catch (Exception e)
               {
                   Console.WriteLine(e.ToString());
               }
               finally
               {
                   concurrentDictionary.TryRemove(guid, out _);
               }
           }, TaskCreationOptions.PreferFairness);
        }


        /// <summary>
        /// 发送
        /// </summary>
        /// <param name="guid"></param>
        /// <param name="data"></param>
        public void Send(Guid guid, MTPPackage package)
        {
            if (package == null)
            {
                return;
            }
            int index = 0;
            if (package.Stream.CanRead)
            {
                package.Stream.Position = 0;
                while (true)
                {
                    byte[] bytes = new byte[MTPConst.MAX_LENGTH];
                    int length = package.Stream.Read(bytes, 0, bytes.Length);
                    bool isEnd = length < MTPConst.MAX_LENGTH;
                    if (isEnd)
                        Array.Resize(ref bytes, length);
                    if (index == 0)
                        Write(Util.BuilderFirestPackage(guid, package.PackageType, package.Header, isEnd, bytes));
                    else
                        Write(Util.BuilderPackage(guid, index, isEnd, bytes));
                    if (isEnd)
                        break;
                    index++;
                };
            }
            else
            {
                throw new IOException("Unreadable");
            }
        }

        public void CloseThrow()
        {
            Close();
            throw new IOException("the connection has been disconnected.");
        }


        /// <summary>
        /// 写入字节
        /// </summary>
        /// <param name="bytes"></param>
        private void Write(ByteWriteBuffer buffer)
        {
            try
            {
                lock (this)
                {
                    socket.Send(buffer.ToBytes());
                }
            }
            catch
            {
                CloseThrow();
            }
        }
        public byte[] ReadBytes(int capacity)
        {
            if (capacity == 0)
            {
                return Array.Empty<byte>();
            }
            byte[] bytes = new byte[capacity];
            int length = socket.Receive(bytes);
            if (length == 0)
            {
                CloseThrow();
            }
            if (length < bytes.Length)
                Array.Resize(ref bytes, length);
            return bytes;
        }

        /// <summary>
        /// 读取字节
        /// </summary>
        /// <param name="capacity"></param>
        /// <returns></returns>
        public ByteReadBuffer Read(int capacity)
        {
            return ByteReadBuffer.Wrap(ReadBytes(capacity));
        }


        public void Close()
        {
            if (!IsClose)
            {
                IsClose = true;
                if (socket.Connected)
                {
                    socket.Shutdown(SocketShutdown.Both);
                    socket.Close();
                }
                dispatch?.Dispose(ClientId);
            }
        }


        public void Dispose()
        {
            if (socket.Connected)
            {
                socket.Disconnect(true);
            }
            try
            {
                socket.Shutdown(SocketShutdown.Both);
            }
            catch
            {
            }
            socket.Close();
            socket.Dispose();
        }
    }
}
